package com.etl.dataflow.executor.communication.service;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.etl.dataflow.common.entity.TaskInfo;
import com.etl.dataflow.common.entity.TaskStatisticInfo;
import com.etl.dataflow.common.exception.EtlException;
import com.etl.dataflow.common.exception.ServiceException;
import com.etl.dataflow.common.response.Code;
import com.etl.dataflow.common.response.R;
import com.etl.dataflow.common.util.JvmInfo;
import com.etl.dataflow.common.util.ResponseUtil;
import com.etl.dataflow.executor.communication.ExecutorMark;
import com.etl.dataflow.executor.communication.TaskRunnerHolder;
import com.etl.dataflow.executor.communication.form.BatchTaskStatisticForm;
import com.etl.dataflow.executor.core.*;
import com.etl.dataflow.executor.entity.ExecutorTaskInfo;
import com.etl.dataflow.executor.util.SchedulerInfoUtil;
import com.sun.management.OperatingSystemMXBean;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.lang.management.ManagementFactory;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author dx
 * @since 2022/4/21
 */
@Slf4j
@Service
public class ExecutorService {

    private static final OperatingSystemMXBean SYSTEM_MX_BEAN = (OperatingSystemMXBean)
            ManagementFactory.getOperatingSystemMXBean();
    /**
     * 通知调度中心更新任务的状态 最大重试次数
     */
    public static final int MAX_RETRY_NUM = 3;
    @Resource
    private SchedulerInfoUtil schedulerInfoUtil;
    @Resource
    private ApplicationContext applicationContext;

    /**
     * 执行任务，并设置回调通知调度中心更新该任务的状态
     *
     * @param taskInfo 任务信息
     * @return R
     */
    public R runTask(TaskInfo taskInfo) {
        log.debug(JSONUtil.toJsonStr(JSONUtil.parse(taskInfo), 2));
        ExecutorTaskInfo executorTaskInfo = new ExecutorTaskInfo();
        BeanUtil.copyProperties(taskInfo, executorTaskInfo);
        // 根据配置的拦截器名称获取各拦截器的实例, 并 set 到任务信息的扩展字段中
        injectInterceptorsForExecutorTaskInfo(taskInfo, executorTaskInfo);

        TaskRunner taskRunner = new TaskRunner(executorTaskInfo);
        AtomicBoolean taskError = new AtomicBoolean(false);
        taskRunner.setOnTaskFinishedListener(() -> {
            if (!taskError.get()) {
                this.notifySchedulerTaskFinished(taskInfo);
            }
            // 任务成功，将任务信息删除
            TaskRunnerHolder.getInstance().removeTaskRunner(taskInfo.getId());
        });
        taskRunner.setOnTaskErrorListener(exception -> {
            taskError.set(true);
            this.notifySchedulerTaskError(taskInfo, exception);
            // 任务失败，将任务信息删除
            TaskRunnerHolder.getInstance().removeTaskRunner(taskInfo.getId());
        });
        // 将任务记录到taskStateHolder中
        TaskRunnerHolder.getInstance().addTaskRunner(taskInfo.getId(), taskRunner);
        taskRunner.run();
        return R.success();
    }

    /**
     * 根据TaskInfo中配置的各个拦截器BeanName从spring容器获取实例
     *
     * @param taskInfo         任务信息
     * @param executorTaskInfo 任务信息的扩展
     */
    private void injectInterceptorsForExecutorTaskInfo(TaskInfo taskInfo, ExecutorTaskInfo executorTaskInfo) {
        String taskLaunchInterceptorName = taskInfo.getTaskLaunchInterceptorBeanName();
        String afterReadInterceptorName = taskInfo.getAfterReadInterceptorBeanName();
        String afterWriteInterceptorName = taskInfo.getAfterWriteInterceptorBeanName();

        AbstractTaskLaunchInterceptor taskLaunchInterceptor = createInterceptorInstance(taskLaunchInterceptorName,
                AbstractTaskLaunchInterceptor.class);
        executorTaskInfo.setTaskLaunchInterceptor(taskLaunchInterceptor);

        AbstractAfterReadInterceptor afterReadInterceptor = createInterceptorInstance(afterReadInterceptorName,
                AbstractAfterReadInterceptor.class);
        executorTaskInfo.setAfterReadInterceptor(afterReadInterceptor);

        AbstractAfterWriteInterceptor afterWriteInterceptor = createInterceptorInstance(afterWriteInterceptorName,
                AbstractAfterWriteInterceptor.class);
        executorTaskInfo.setAfterWriteInterceptor(afterWriteInterceptor);
    }

    /**
     * 根据拦截器的BeanName创建拦截器实例
     *
     * @param interceptorName  拦截器BeanName
     * @param interceptorClass 拦截器Class
     * @param <T>              拦截器类型
     * @return 拦截器实例，拦截器BeanName为空时返回null
     */
    private <T extends Interceptor> T createInterceptorInstance(String interceptorName, Class<T> interceptorClass) {
        if (StrUtil.isNotEmpty(interceptorName)) {
            T interceptor;
            try {
                interceptor = applicationContext.getBean(interceptorName, interceptorClass);
            } catch (NoSuchBeanDefinitionException e) {
                String errorMsg = String.format(Code.NO_SUCH_INTERCEPTOR_FOUND.getMsg(), interceptorName);
                throw new ServiceException(Code.NO_SUCH_INTERCEPTOR_FOUND.getCode(), errorMsg);
            }
            checkIsInterceptorSupportedMultiInstance(interceptorName, interceptor);
            return interceptor;
        }
        return null;
    }

    private void checkIsInterceptorSupportedMultiInstance(String launchInterceptorName, Interceptor interceptor) {
        Scope scope = interceptor.getClass().getAnnotation(Scope.class);
        boolean isNotPrototype = (scope != null) && (!"prototype".equals(scope.value()));
        if (scope == null || isNotPrototype) {
            throw new EtlException(String.format("请为拦截器%s添加@Scope(\"prototype\")注解来支持多实例", launchInterceptorName));
        }
    }

    private void notifySchedulerTaskError(TaskInfo taskInfo, Exception e) {
        boolean success = tryNotifySchedulerTaskError(taskInfo, e);
        if (success) {
            return;
        }
        // 如果失败了再尝试三次
        for (int i = 0; i < MAX_RETRY_NUM; i++) {
            if (tryNotifySchedulerTaskError(taskInfo, e)) {
                return;
            }
        }
    }

    private boolean tryNotifySchedulerTaskError(TaskInfo taskInfo, Exception e) {
        String subUrl = String.format("/state/error/%s/%s/%s", ExecutorMark.getInstance().getUuid(),
                taskInfo.getId(), taskInfo.getTaskRecordId());
        try (HttpResponse httpResponse = HttpUtil
                .createPost(schedulerInfoUtil.getSchedulerUrl() + subUrl)
                .form("errorMsg", e.getMessage())
                .execute()) {
            String body = httpResponse.body();
            log.debug("通知调度中心任务状态为出现致命错误，返回结果：" + body);
            return ResponseUtil.isSuccess(body) || ResponseUtil.codeEquals(body, Code.TASK_STATE_UNMODIFIABLE);
        }
    }

    /**
     * 通知调度中心更新任务状态为正常结束
     */
    private void notifySchedulerTaskFinished(TaskInfo taskInfo) {
        boolean success = tryNotifySchedulerTaskFinished(taskInfo);
        if (success) {
            return;
        }
        // 如果失败了再尝试三次
        for (int i = 0; i < MAX_RETRY_NUM; i++) {
            if (tryNotifySchedulerTaskFinished(taskInfo)) {
                return;
            }
        }
    }

    private boolean tryNotifySchedulerTaskFinished(TaskInfo taskInfo) {
        String subUrl = String.format("/state/finished/%s/%s/%s", ExecutorMark.getInstance().getUuid(),
                taskInfo.getId(), taskInfo.getTaskRecordId());
        try (HttpResponse response = HttpUtil
                .createGet(schedulerInfoUtil.getSchedulerUrl() + subUrl)
                .execute()) {
            String body = response.body();
            JSONObject jsonObject = JSONUtil.parseObj(body);
            boolean success = false;
            if (ResponseUtil.isSuccess(jsonObject) && jsonObject.getJSONObject("data") != null) {
                JSONObject data = jsonObject.getJSONObject("data");
                Boolean result = data.getBool("result");
                if (result != null && result) {
                    log.debug("taskRecordId: " + taskInfo.getTaskRecordId() + "，通知调度中心任务状态为正常结束成功，返回结果：" + body);
                    success = true;
                }
            } else if (ResponseUtil.codeEquals(jsonObject, Code.TASK_STATE_UNMODIFIABLE)) {
                // 调度器已告知任务状态不可更改，重试也是在做无用功，不如返回成功避免多次重试
                success = true;
            }
            if (!success) {
                log.debug("taskRecordId: " + taskInfo.getTaskRecordId() + "，通知调度中心任务状态为正常结束失败，返回结果：" + body);
            }
            return success;
        }
    }

    public R getTaskStatistic(String taskId) {
        Optional<TaskRunner> taskRunner = TaskRunnerHolder.getInstance().getTaskRunner(taskId);
        if (!taskRunner.isPresent()) {
            throw new ServiceException(Code.TASK_NOT_FOUND_CANNOT_STAT);
        }
        return R.success().put("statistic", collectStatistic(taskRunner.get()));
    }

    public R getTaskStatisticBatch(BatchTaskStatisticForm form) {
        List<String> taskIds = form.getTaskIds();
        TaskRunnerHolder taskRunnerHolder = TaskRunnerHolder.getInstance();
        Map<String, TaskStatisticInfo.Dto> statMap = new HashMap<>((int) Math.ceil(taskIds.size() * 1.4));
        for (String taskId : taskIds) {
            Optional<TaskRunner> taskRunner = taskRunnerHolder.getTaskRunner(taskId);
            if (taskRunner.isPresent()) {
                TaskStatisticInfo.Dto taskStatInfoDto = collectStatistic(taskRunner.get());
                statMap.put(taskId, taskStatInfoDto);
            }
        }
        return R.success().put("statistic", statMap);
    }

    private TaskStatisticInfo.Dto collectStatistic(TaskRunner taskRunner) {
        TaskStatisticInfo statisticInfo = taskRunner.getTaskStatisticInfo();
        return new TaskStatisticInfo.Dto()
                .setSrcRowNumber(statisticInfo.getSrcRowNumber().get())
                .setRedRowNumber(statisticInfo.getRedRowNumber().get())
                .setProcessedRowNumber(statisticInfo.getProcessedRowNumber().get())
                .setWrittenRowNumber(statisticInfo.getWrittenRowNumber().get())
                .setReadBytesPerSecond(statisticInfo.getReadBytesPerSecond().get())
                .setReadRowsPerSecond(statisticInfo.getReadRowsPerSecond().get())
                .setWriteBytesPerSecond(statisticInfo.getWriteBytesPerSecond().get())
                .setWriteRowsPerSecond(statisticInfo.getWriteRowsPerSecond().get())
                .setWriteBufferBlockingTimePer10Kr(statisticInfo.getWriteBufferBlockingTimePer10Kr().longValue())
                .setReadBufferBlockingTimePer10Kr(statisticInfo.getReadBufferBlockingTimePer10Kr().longValue());
    }

    public R terminateTask(String taskId) {
        Optional<TaskRunner> taskRunner = TaskRunnerHolder.getInstance().getTaskRunner(taskId);
        if (!taskRunner.isPresent()) {
            throw new ServiceException(Code.TASK_NOT_FOUND_CANNOT_TERMINATE);
        }
        taskRunner.get().shutdown();
        return R.success();
    }

    public R executorJvmInfo() {
        return R.success().put("jvmInfo", JvmInfo.getJvmInfo().toString());
    }

    public R getRunningTasks() {
        List<TaskRunner> allTaskRunner = TaskRunnerHolder.getInstance().getAllTaskRunner();
        List<TaskInfo> taskInfos = new ArrayList<>();
        ExecutorTaskInfo executorTaskInfo;
        for (TaskRunner taskRunner : allTaskRunner) {
            executorTaskInfo = taskRunner.getExecutorTaskInfo();
            Optional.ofNullable(executorTaskInfo.getTaskParam().getExtensionMap()).ifPresent(it -> it.remove("writerFactory"));
            taskInfos.add(executorTaskInfo);
        }
        return R.success().put("data", taskInfos);
    }

    public R getExecutorRuntimeInfo() {
        final long GB = 1024 * 1024 * 1024;
        double systemCpuLoad = SYSTEM_MX_BEAN.getSystemCpuLoad() * 100;
        double totalMemory = 1.0 * SYSTEM_MX_BEAN.getTotalPhysicalMemorySize() / GB;
        double freeMemory = 1.0 * SYSTEM_MX_BEAN.getFreePhysicalMemorySize() / GB;
        int taskNumber = TaskRunnerHolder.getInstance().getRunningTaskNumber();
        return R.success()
                .put("sysCpu", twoDecimal(systemCpuLoad))
                .put("totalMemory", twoDecimal(totalMemory))
                .put("freeMemory", twoDecimal(freeMemory))
                .put("taskNumber", taskNumber);
    }

    public double twoDecimal(double doubleValue) {
        BigDecimal bigDecimal = new BigDecimal(doubleValue).setScale(2, RoundingMode.HALF_UP);
        return bigDecimal.doubleValue();
    }
}
